package com.mk.rocket;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;

import java.nio.charset.StandardCharsets;
import java.util.List;

/**
 * @Author makai
 * @Description: TODO
 * @Date 2024/1/31 20:28
 */
@org.springframework.context.annotation.Configuration
public class RocketMqConfiguration {

    @Value("${rocketmq.namesrv.addr}")
    private String namesrvAddr;


//    @Bean
//    public DefaultMQProducer defaultMQProducer(){
//        DefaultMQProducer producer = new DefaultMQProducer("test-producer-group");
//        producer.setNamesrvAddr(namesrvAddr);
//        return producer;
//    }


    @Bean(name = "defaultMQProducer")
    public TransactionMQProducer transactionMQProducer(){
        TransactionMQProducer producer = new TransactionMQProducer("test-producer-group");
        producer.setNamesrvAddr(namesrvAddr);
//        //事务处理
//        producer.setTransactionListener(new TransactionListener() {
//            //sdk提交完事务到broker后会调用此方法，业务中需要进行本地事务的提交
//            @Override
//            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
//                LocalTransactionState state = LocalTransactionState.COMMIT_MESSAGE;
//                return state;
//            }
//            //broker没有收到 executeLocalTransaction 结果时会调用此方法检查本地事务的提交情况
//            @Override
//            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
//                LocalTransactionState state = LocalTransactionState.COMMIT_MESSAGE;
//                return state;
//            }
//        });
        return producer;
    }


    @Bean(initMethod = "start")
    public DefaultMQPushConsumer defaultMQPushConsumer() throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test-consumer-group");
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.subscribe("test-topic", "*");
        consumer.subscribe("test-delay-topic", "*");
        consumer.subscribe("test-transaction-topic", "*");
        consumer.subscribe("test-orderly-topic", "*");
        consumer.subscribe("transaction-topic", "*");
        //消费位点（新的消费组上线时从队列的第一个消息开始消费）
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.registerMessageListener(new ConsumerListener());
        consumer.setMaxReconsumeTimes(10); //消费失败时最大消费次数，默认值：顺序消息无限重试；普通消息：16次。最终失败后会进入到死信队列
        return consumer;
    }

    /**
     * 测试一个进程中创建多个消费者组订阅相同的topic
     * @return
     * @throws MQClientException
     */
//    @Bean(initMethod = "start")
    public DefaultMQPushConsumer defaultMQPushConsumer2() throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test-consumer-group");
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.subscribe("test-topic", "*");
        consumer.subscribe("test-delay-topic", "*");
        consumer.subscribe("test-transaction-topic", "*");
        consumer.subscribe("test-orderly-topic", "*");
        consumer.subscribe("transaction-topic", "*");
        //消费位点（新的消费组上线时从队列的第一个消息开始消费）
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.registerMessageListener(new MessageListenerConcurrently(){
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                System.out.println("c1收到消息:"+new String(msgs.get(0).getBody(), StandardCharsets.UTF_8));
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.setMaxReconsumeTimes(10); //消费失败时最大消费次数，默认值：顺序消息无限重试；普通消息：16次。最终失败后会进入到死信队列
        return consumer;
    }

}
